package com.qf.point.consumer;


import com.alibaba.fastjson.JSON;
import com.qf.api.point.dto.PointDTO;
import com.qf.common.core.entity.Points;
import com.qf.common.core.utils.UserUtils;
import com.qf.common.redis.utils.RedisUtil;
import com.qf.common.rocket.param.PointMessage;
import com.qf.point.service.PointsService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author zdl
 * @date 2025/4/15
 */
@Component
@Slf4j
@RocketMQMessageListener(topic = "USER_TOPIC1", consumerGroup = "USER_GROUP")
public class PointConsumer implements RocketMQListener<MessageExt> , RocketMQPushConsumerLifecycleListener {
    @Resource
    private PointsService pointsService;

    @Resource
    private RedisUtil redisUtil;

    // 消费者消费消息后，需要给broker返回ack消息
    // 若消息处理成功，但是返回ack时，出现网络抖动
    // broker可能没有办法收到ack，超时后，消费者重新消费消息
    // 进行幂等性处理，防止消息的重复消费
    @Override
    public void onMessage(MessageExt s) {
        log.info("接收消息:{}", s);
        String keys = s.getKeys();
        if (redisUtil.keyExist("reconsume:" + keys)) {
            log.info("消息消费过");
            return;
        }
        String message = new String(s.getBody());
        PointDTO pointDTO = JSON.parseObject(message, PointDTO.class);
        Points points = new Points();
        BeanUtils.copyProperties(pointDTO, points);
        // 如果送积分失败，进行积分补偿
        pointsService.addPoint(points);
        redisUtil.setCacheObject("reconsume:" + keys, "1", 60, TimeUnit.SECONDS);
        log.info("add point success");
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.setMaxReconsumeTimes(0);
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt ext : list) {
                    String keys = ext.getKeys();
                    if (redisUtil.keyExist("reconsume:" + keys)) {
                        log.info("消息消费过");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    String message = new String(ext.getBody());
                    PointDTO pointDTO = JSON.parseObject(message, PointDTO.class);
                    Points points = new Points();
                    BeanUtils.copyProperties(pointDTO, points);
                    // 如果送积分失败，进行积分补偿
                    try {
                        pointsService.addPoint(points);
                        redisUtil.setCacheObject("reconsume:" + keys, "1", 60, TimeUnit.SECONDS);
                    } catch (Exception e) {
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
